package com.winit.schedule.job.tracker.repository.mysql;

import java.util.*;

import javax.annotation.Resource;

import com.winit.schedule.core.repository.entity.*;
import com.winit.schedule.core.repository.query.WhiteListQuery;
import com.winit.schedule.job.tracker.repository.mysql.dao.*;
import org.eclipse.jetty.util.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import com.winit.schedule.core.model.AssignedTask;
import com.winit.schedule.core.model.JobID;
import com.winit.schedule.core.model.TaskID;
import com.winit.schedule.core.repository.constants.JobStatus;
import com.winit.schedule.core.repository.constants.LogType;
import com.winit.schedule.core.repository.constants.TaskItemStatus;
import com.winit.schedule.core.repository.constants.TaskStatus;
import com.winit.schedule.core.repository.query.TaskItemQuery;
import com.winit.schedule.core.task.ScheduleTaskDefine;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;

@Service("scheduleDataManager")
@Transactional
public class MysqlScheduleDataManager extends MysqlLogDataManager implements ScheduleDataManager {

    private static final Logger logger = LoggerFactory.getLogger(MysqlScheduleDataManager.class);

    public static final Set<String> IP_ADDRESS = new ConcurrentHashSet<String>();

    public static final Object cacheLock = new Object();

    @Resource
    private MysqlTaskDao            taskDao;

    @Resource
    private MysqlTaskDefineDao      taskDefineDao;

    @Resource
    private MysqlTaskItemDao        taskItemDao;

    @Resource
    private MysqlTaskTrackerDao     taskTrackerDao;

    @Resource
    private MysqlJobDao             jobDao;

    @Resource
    private MysqlTaskRunRecordDao   taskRunRecordDao;

    @Resource
    private MysqlQuartzSchedulerDao quartzSchedulerDao;

    @Resource
    private MysqlQuartzJobDao       quartzJobDao;

    @Override
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public TaskEntity getTaskEntityById(String taskId) {
        return taskDao.getTaskById(taskId);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public TaskItemEntity findAndFetchTaskItemByTaskId(String taskTrackerUuid, String taskId, long nextRunTime,
                                                       int fetchNum) {
        TaskItemEntity taskItem = null;
        long currentTime = Clock.getSystemTime();
        taskItem = taskItemDao.findRunnableTaskItemByTaskId(taskId, currentTime);
        int updateNum = 0;
        if (null != taskItem) {
            updateNum = taskItemDao.fetchRunnableTaskItemByTaskId(taskItem.getTaskItemId(),
                taskTrackerUuid,
                currentTime,
                nextRunTime);
        }
        if (updateNum > 0) {
            taskItem.setLastFetchTime(currentTime);
            taskItem.setLastFetchTaskTrackerUuid(taskTrackerUuid);
            taskItem.setNextRunTime(nextRunTime);
            return taskItem;
        } else {
            return null;
        }
    }

    @Override
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public long getTaskItemNumberOfPerTasktracker(String taskId) {
        int fetchNum = 1;
        String[] arr = StringUtils.delimitedListToStringArray(taskId, "$");
        long activiteTaskTrackerNum = taskTrackerDao.queryTaskTackerNumberByTaskId(arr[0], arr[1]);
        long taskItemNum = taskItemDao.queryTaskItemNumByTaskId(taskId, Clock.getSystemTime());

        int val = (int) (taskItemNum / activiteTaskTrackerNum);

        if (taskItemNum % activiteTaskTrackerNum != 0) {
            val = val + 1;
        }

        return val > fetchNum ? val : fetchNum;
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void recoveryPushFailureTaskItem(long timeout) {
        Map<String, Object> params = new HashMap<String, Object>();
        params.put("timeout", Clock.getSystemTime() - timeout);
        params.put("taskItemStatus", TaskItemStatus.RUN_STATUS);
        taskItemDao.recoveryPushFailureTaskItem(params);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void removeTaskTrackerAndRunnerForTimeout(long heartbeatTimeout) {

        long timeout = Clock.getSystemTime() - heartbeatTimeout;
        List<TaskTrackerEntity> taskTrackers = taskTrackerDao.queryTaskTrackerForTimeout(timeout);

        if (null != taskTrackers && taskTrackers.size() > 0) {
            taskTrackerDao.removeTaskTrackerForTimeout(timeout);

            List<String> uuids = new ArrayList<String>();
            for (TaskTrackerEntity taskTracker : taskTrackers) {
                taskDefineDao.removeTaskDefineByTaskTrackerUuid(taskTracker.getTaskTrackerUuid());
                uuids.add(taskTracker.getTaskTrackerUuid());
            }

            Map<String, Object> params = new HashMap<String, Object>();
            params.put("taskTrackerUuid", uuids);
            params.put("taskItemStatus", TaskItemStatus.RUN_STATUS);
            taskItemDao.removeTaskRunnerByTaskTrackerUuids(params);
        }

    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void removeTaskTrackerAndRunnerByUuid(String taskTrackerUuid) {
        taskTrackerDao.removeTaskTracker(taskTrackerUuid);
        taskDefineDao.removeTaskDefineByTaskTrackerUuid(taskTrackerUuid);

        Map<String, Object> params = new HashMap<String, Object>();
        params.put("taskTrackerUuid", taskTrackerUuid);
        params.put("taskItemStatus", TaskItemStatus.RUN_STATUS);
        taskItemDao.removeTaskRunnerByTaskTrackerUuid(params);

        logger(taskTrackerUuid, LogType.TASKTRACKER_DISCONNECT);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void updateTaskItemStatusByTaskRunnerUuid(TaskItemEntity taskItem) {
        String taskItemId = String.valueOf(taskItem.getTaskItemId());
        synchronized (taskItemId) {
            Map<String, Object> params = new HashMap<String, Object>();
            // params.put("taskRunnerUuid",
            // taskItem.getTaskRunner().getTaskRunnerUuid());
            params.put("status", taskItem.getTaskItemStatus());
            params.put("taskItemId", taskItem.getTaskItemId());
            params.put("lastFetchTime", taskItem.getLastFetchTime());
            params.put("lastFetchTaskTrackerUuid", taskItem.getLastFetchTaskTrackerUuid());
            taskItemDao.updateTaskItemStatusByTaskRunnerUuid(params);
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void updateTaskRunnerForTaskItem(TaskItemEntity taskItemEntity) {
        if (null != taskItemEntity) {
            String taskItemId = String.valueOf(taskItemEntity.getTaskItemId());
            synchronized (taskItemId) {
                taskItemDao.updateTaskRunnerForTaskItem(taskItemEntity);
            }
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public TaskTrackerEntity getTaskTrackerByUuId(String uuid) {
        TaskTrackerEntity taskTracker = taskTrackerDao.getTaskTrackerByUuid(uuid);

        List<TaskDefineEntity> taskDefines = taskDefineDao.findTaskDefineByTaskTrackerUuid(uuid);

        if (null != taskTracker && null != taskDefines && taskDefines.size() > 0) {
            taskTracker.getTaskDefines().addAll(taskDefines);
            return taskTracker;
        } else {
            taskTrackerDao.removeTaskTracker(uuid);
            taskDefineDao.removeTaskDefineByTaskTrackerUuid(uuid);
            return null;
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public synchronized TaskTrackerEntity saveOrUpdateTaskTrackerAndUpdateTaskInterfaceType(TaskTrackerEntity entity) {
        TaskTrackerEntity taskTracker = taskTrackerDao.getTaskTrackerByUuid(entity.getTaskTrackerUuid());
        if (taskTracker != null) {
            taskTrackerDao.updateTaskTracker(entity);
        } else {
            taskTrackerDao.saveTaskTracker(entity);

            for (ScheduleTaskDefine taskDefine : entity.getTaskDefines()) {
                TaskDefineEntity defineEntity = new TaskDefineEntity();
                BeanUtils.copyProperties(taskDefine, defineEntity);
                defineEntity.setTaskTrackerUuid(entity.getTaskTrackerUuid());
                taskDefineDao.saveTaskDefine(defineEntity);

                TaskEntity task = taskDao.getTaskById(taskDefine.getTaskId());
                if (null == task) {
                    task = new TaskEntity();
                    task.setInterfaceType(taskDefine.getInterfaceType());
                    task.setTaskId(new TaskID(taskDefine.getTaskId()));
                    task.setTaskGroup(taskDefine.getGroup());
                    task.setTaskName(taskDefine.getTaskName());
                    taskDao.saveTask(task);
                } else {
                    taskDao.updateTaskInterfaceType(taskDefine.getTaskId(), taskDefine.getInterfaceType());
                }
            }
        }

        return entity;
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public TaskTrackerEntity updateHeartbeatTime(TaskTrackerEntity entity) {
        taskTrackerDao.updateHeartbeatTime(entity);
        return entity;
    }

    @Override
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public List<TaskID> queryAllStopTaskIDs() {
        return taskDao.queryAllStopTaskIDs();
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public synchronized void completeTask(TaskRunnerEntity taskRunner, AssignedTask task) {
        taskDao.updateTaskStatusByTaskId(task.getTaskId().getTaskId(), TaskStatus.COMPLETE);

        // updateTaskItemStatusByTaskRunnerUuid(taskRunner, task.getTaskItems(),
        // TaskItemStatus.COMPLETE);

        JobID jobId = task.getTaskId().getJobId();
        if (null != jobId) {
            jobDao.updateJobCompleteStatus(jobId.getJobId(), JobStatus.COMPLETE, TaskStatus.COMPLETE);
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void saveTaskRunRecord(TaskRunRecordEntity taskRunRecord) {
        TaskRunRecordEntity record = taskRunRecordDao.getTaskRunRecordById(taskRunRecord.getTaskRunnerUuid());
        if (null != record) {
            taskRunRecordDao.updateTaskRunRecord(taskRunRecord);
        } else {
            taskRunRecordDao.saveTaskRunRecord(taskRunRecord);
        }

    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void updateQuartzScheduler(List<QuartzSchedulerEntity> schedulers) {
        if (null != schedulers && schedulers.size() > 0) {
            QuartzSchedulerEntity oldSchedulerEntity = null;
            QuartzJobEntity oldJobEntity = null;
            long currentTime = Clock.getSystemTime();
            for (QuartzSchedulerEntity schedulerEntity : schedulers) {
                oldSchedulerEntity = quartzSchedulerDao.getSchedulerByCompositeKey(schedulerEntity);
                if (null != oldSchedulerEntity) {
                    schedulerEntity.setHeartbeatTime(currentTime);
                    quartzSchedulerDao.updateSchedulerStatusAndHeartbeatTime(schedulerEntity);
                } else {
                    schedulerEntity.setRegisterTime(currentTime);
                    schedulerEntity.setHeartbeatTime(currentTime);
                    quartzSchedulerDao.saveScheduler(schedulerEntity);
                }
                List<QuartzJobEntity> jobs = schedulerEntity.getJobs();
                if (null != jobs) {
                    for (QuartzJobEntity jobEntity : jobs) {
                        jobEntity.setLastStatusUpdateTime(currentTime);
                        oldJobEntity = quartzJobDao.getJobByCompositeKey(jobEntity);
                        if (null != oldJobEntity) {
                            quartzJobDao.updateJobStatus(jobEntity);
                        } else {
                            quartzJobDao.saveJob(jobEntity);
                        }
                    }
                }
            }
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void deleteQuartzScheduler(String taskTrackerUuid) {
        quartzSchedulerDao.deleteSchedulerByTaskTrackerUuid(taskTrackerUuid);
        quartzJobDao.deleteJobsByTaskTrackerUuid(taskTrackerUuid);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void cleanQuartzCommand(String taskTrackerUuid) {
        quartzSchedulerDao.cleanCommandByTaskTrackerUuid(taskTrackerUuid);
        quartzJobDao.cleanCommandByTaskTrackerUuid(taskTrackerUuid);
    }

    @Override
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public List<QuartzJobEntity> queryJobCommnads(String taskTrackerUuid) {
        return quartzJobDao.queryJobCommnads(taskTrackerUuid);
    }

    @Override
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public List<QuartzSchedulerEntity> querySchedulerCommnads(String taskTrackerUuid) {
        return quartzSchedulerDao.querySchedulerCommnads(taskTrackerUuid);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void cleanQuartzTimeout(long timeout) {
        long time = Clock.getSystemTime() - timeout;
        quartzSchedulerDao.deleteSchedulerTimeout(time);
        quartzJobDao.deleteJobTimeout(time);
    }

    @Override
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public List<TaskItemEntity> queryTaskItemsByTimeout(TaskItemQuery query) {
        return taskItemDao.queryTaskItemsByTimeout(query);
    }

    @Override
    public int deleteLogByLogtime(long logTime) {
        return logDao.deleteLogByLogtime(logTime);
    }

    @Override
    public int deleteRunRecordByStarttime(long runStartTime) {
        return taskRunRecordDao.deleteRunRecordByStarttime(runStartTime);
    }

    @Override
    public void saveScheduleTaskDefine(ScheduleTaskDefine taskDefine, String taskTrackerUUID) {
        TaskDefineEntity defineEntity = new TaskDefineEntity();
        BeanUtils.copyProperties(taskDefine, defineEntity);
        defineEntity.setTaskTrackerUuid(taskTrackerUUID);
        taskDefineDao.saveTaskDefine(defineEntity);

        TaskEntity task = taskDao.getTaskById(taskDefine.getTaskId());
        if (null == task) {
            task = new TaskEntity();
            task.setInterfaceType(taskDefine.getInterfaceType());
            task.setTaskId(new TaskID(taskDefine.getTaskId()));
            task.setTaskGroup(taskDefine.getGroup());
            task.setTaskName(taskDefine.getTaskName());
            taskDao.saveTask(task);
        } else {
            taskDao.updateTaskInterfaceType(taskDefine.getTaskId(), taskDefine.getInterfaceType());
        }
    }

    @Override
    public boolean containBlackList(String ipaddress) {
        if (null == IP_ADDRESS || IP_ADDRESS.size() == 0) {
            return false;
        } else {
            return IP_ADDRESS.contains(ipaddress);
        }
    }

    @Override
    public boolean resetBlackListCache(String blackList) {
        try {
            synchronized (cacheLock) {
                if (null != blackList && blackList.trim().length() > 0) {

                    String[] datas = StringUtils.tokenizeToStringArray(blackList,",",true,true);

                    Set<String> newBlankList = new HashSet<String>(Arrays.asList(datas));

                    List<String> removes = new ArrayList<String>();

                    if (IP_ADDRESS.size() > 0) {
                        Iterator<String> iter = IP_ADDRESS.iterator();
                        String val = null;
                        while (iter.hasNext()) {
                            val = iter.next();
                            if (newBlankList.contains(val)) {
                                newBlankList.remove(val);
                            } else {
                                removes.add(val);
                            }
                        }
                    }

                    if (removes.size() > 0) {
                        IP_ADDRESS.removeAll(removes);
                    }
                    if (newBlankList.size() > 0) {
                        IP_ADDRESS.addAll(newBlankList);
                    }
                } else {
                    IP_ADDRESS.clear();
                }
            }
            return true;
        } catch (Throwable e) {
            logger.error("", e);
            IP_ADDRESS.clear();
            return false;
        }

    }

    @Override public List<TaskID> queryAllTaskIdByTaskTracker(String taskGroup) {
        if (StringUtils.isEmpty(taskGroup)) {
            return Collections.emptyList();
        } else {
            return taskDao.queryAllTaskIdByTaskTracker(taskGroup);
        }
    }

}
